package com.qqt.csr.ixport.service;

import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.ixport.entity.ImportTopicConfig;
import com.qqt.csr.ixport.model.ChunkMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

/**
 * @Author: yang hao
 * @Date: 2025/2/6 14:47
 * @Description: 文件分块生产者
 **/
@Slf4j
@Service
public class FileChunkProducer {

    /**
     * RocketMQTemplate
     */
    private final RocketMQTemplate rocketMQTemplate;


    /**
     * 构造方法
     * @param rocketMQTemplate rocketMQTemplate
     */
    public FileChunkProducer(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    /**
     * 发送分片
     * @param config topic配置
     * @param message 消息体
     */
    public boolean sendChunk(ImportTopicConfig config, ChunkMessage message) {
        String topic = buildTopic(config);
        if(StringUtils.isBlank(topic)){
            return false;
        }
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(JsonUtil.toJson(message)).build());
        log.info("【send file chunk】sendResult={}", JsonUtil.toJson(sendResult));
        return sendResult.getSendStatus() == SendStatus.SEND_OK;
    }

    /**
     * 构建topic
     * @param config 导入主题配置
     * @return topic
     */
    public String buildTopic(ImportTopicConfig config) {
        if(config == null || StringUtils.isBlank(config.getTopic())){
            return StringUtils.EMPTY;
        }
        if(StringUtils.isBlank(config.getTag())){
            return config.getTopic();
        }
        return String.format("%s:%s", config.getTopic(), config.getTag());
    }

}
